package org.medusa.db.handlersocket;

import java.util.HashMap;
import java.util.Map;
import java.util.Vector;

import org.medusa.core.Core;
import org.medusa.db.SyncDBConnector;
import org.medusa.log.ILog;
import org.medusa.network.tcp.TcpSyncLineConnector;
import org.medusa.packet.blob.LinePacket;

import com.sun.corba.se.impl.orbutil.concurrent.Mutex;

public class HSConnector extends SyncDBConnector {
	// hs最大发送缓冲区大小
	// private static final int HANDLERSOCKET_SNDBUF_SIZE = 1677216;

	// info
	private String name; // host:port:1 //:1是为了处理一个数据库主机有多个连接的情况
	private ILog log;

	// auth
	private String host;
	private int port;
	private String port_passwd;

	// connector
	private TcpSyncLineConnector conn;

	// thread_safe
	private boolean thread_safe;
	private Mutex mutex;

	// index
	int index_max;
	private int pid_counter;
	private Map<Integer, String> pid_dtc_link;
	private Map<String, Integer> dtc_pid_link;
	private Map<Integer, String[]> pid_columns_link; // pid-查询的列

	public HSConnector(int index_max, String name, ILog log, boolean thread_safe) {
		this.name = name;
		this.log = log;

		this.host = "localhost";
		this.port = 9999;
		this.port_passwd = "";

		this.conn = new TcpSyncLineConnector(8192000, 1024000);

		this.thread_safe = thread_safe;
		this.mutex = new Mutex();

		this.index_max = index_max;
		this.pid_counter = 0;
		this.pid_dtc_link = new HashMap<Integer, String>();
		this.dtc_pid_link = new HashMap<String, Integer>();
		this.pid_columns_link = new HashMap<Integer, String[]>();
	}

	public HSConnector(String name, ILog log, boolean thread_safe) {
		this(Integer.MAX_VALUE, name, log, thread_safe);
	}

	public Error Connect(String host, int port, String passwd) {
		Error err = null;
		if (this.conn.IsConnected()) {
			err = new Error("Already connected to server;");
			this.Error(err);
			return err;
		} else {
			this.host = host;
			this.port = port;
			this.port_passwd = passwd;
			err = this.connect();
		}
		return err;
	}

	@Override
	public boolean IsConnected() {
		return this.conn.IsConnected();
	}

	@Override
	public InsertResult Insert(String dbName, String tableName,
			String indexName, String[] columns, Object[] vals) {

		this.Debug("Insert called");
		IndexResult ires = null;
		InsertResult result = new InsertResult();
		CommandResponse response = null;

		int data_num = vals.length;
		String data = this.join_hs_field(vals);

		if (!this.conn.IsConnected()) {
			result.Error("have no connected to mysql!");
			this.Error(result.ErrorMessage());
			return result;
		}

		this.lock();
		// ------------------------

		while (response == null) {

			ires = this.open_index(dbName, tableName, indexName,
					this.join_string(columns, ","));

			if (ires.Error() != null) {
				result.Error(ires.ErrorMessage());
				this.Error(ires.ErrorMessage());
			}

			String cmd = String.format("%d\t%s\t%d\t%s\n", ires.PID(), "+",
					data_num, data);
			this.Stack("insert cmd:[" + cmd + "]");
			// ------------------------
			response = this.run_command(cmd);
			if (response == null) {
				this.reconnect();
				continue;
			}
			if (response.result_code == 1) {
				result.Set("INSERT: Data Exists:" + response.data[1], -2);
				this.Warning(result.ErrorMessage());
			} else if (response.result_code != 0) {
				result.Set("Error Inserting Data:" + response.data[1], -1);
				this.Error(result.ErrorMessage());
			} else {
				if (response.data.length >= 2) {// 存在自增id
					result.IncrementID(Integer.parseInt(response.data[1].trim()));
					this.Debug("Insert return, last insert id:",
							result.IncrementID()); // 自增id
				} else { // 不存在自增id
					result.IncrementID(0);
					this.Debug("Insert return, have no increment field.");
				}
			}
			Core.Gosched();
		}// while
			// ------------------------
		this.unlock();
		this.Debug("Insert finish, err:", result.ErrorMessage());
		return result;
	}

	@Override
	public ModifyResult Modify(String dbName, String tableName,
			String indexName, String oper, Object[] keys, String[] columns,
			Object[] newvals, int limit, int skip) {
		this.Debug("Modify called");
		IndexResult ires = null;
		CommandResponse response = null;
		ModifyResult result = new ModifyResult();

		int key_num = keys.length;
		String key_vals = join_hs_field(keys);
		String new_vals = this.join_hs_field(newvals);

		if (!this.conn.IsConnected()) {
			result.Error("have no connected to mysql!");
			this.Error(result.ErrorMessage());
			return result;
		}

		this.lock();
		// ------------------------
		while (response == null) {
			ires = this.open_index(dbName, tableName, indexName,
					this.join_string(columns, ","));
			if (ires.Error() != null) {
				result.Error(ires.ErrorMessage());
				this.Error(ires.ErrorMessage());
			}

			String cmd = String.format("%d\t%s\t%d\t%s\t%d\t%d\t%s\t%s\n",
					ires.PID(), oper, key_num, key_vals, limit, skip, "U",
					new_vals);

			this.Stack("modify cmd:[" + cmd + "]");
			response = this.run_command(cmd);
			if (response == null) {
				this.reconnect();
				continue;
			}
			if (response.result_code == 1) {
				result.Set("Error Something:" + response.data[1].trim(), 0);
				this.Error(result.ErrorMessage());
			} else {
				result.AffectedRows(Integer.parseInt(response.data[1].trim()));
				this.Debug("Modify return, affected rows:",
						result.AffectedRows());
			}

			Core.Gosched();
		}// while
			// ------------------------
		this.unlock();
		this.Debug("Modify finish, err:", result.ErrorMessage());
		return result;
	}

	@Override
	public ModifyResult Remove(String dbName, String tableName,
			String indexName, String oper, Object[] keys, int limit, int skip) {
		this.Debug("Remove called");
		IndexResult ires = null;
		CommandResponse response = null;
		ModifyResult result = new ModifyResult();

		int key_num = keys.length;
		String ket_vals = join_hs_field(keys);

		if (!this.conn.IsConnected()) {
			result.Error("have no connected to mysql!");
			this.Error(result.ErrorMessage());
			return result;
		}

		this.lock();
		// ------------------------
		while (response == null) {
			ires = this.open_index(dbName, tableName, indexName, "");
			if (ires.Error() != null) {
				result.Error(ires.ErrorMessage());
				this.Error(ires.ErrorMessage());
			}

			String cmd = String.format("%d\t%s\t%d\t%s\t%d\t%d\t%s\t\n",
					ires.PID(), oper, key_num, ket_vals, limit, skip, "D");

			this.Stack("modify cmd:[" + cmd + "]");
			response = this.run_command(cmd);
			if (response == null) {
				this.reconnect();
				continue;
			}
			if (response.result_code == 1) {
				result.Set("Error Something:" + response.data[1].trim(), 0);
				this.Error(result.ErrorMessage());
			} else {
				result.AffectedRows(Integer.parseInt(response.data[1].trim()));
				this.Debug("Remove return, affected rows:",
						result.AffectedRows());
			}
			Core.Gosched();
		}// while
			// ------------------------
		this.unlock();
		this.Debug("Remove finish, err:", result.ErrorMessage());
		return result;
	}

	@Override
	public QueryResult Query(String dbName, String tableName, String indexName,
			String oper, Object[] keys, String[] columns, int limit, int skip) {
		this.Debug("Query called");

		IndexResult ires = null;
		CommandResponse response = null;
		QueryResult result = new QueryResult();
		int key_num = keys.length;
		String key_vals = join_hs_field(keys);

		if (!this.conn.IsConnected()) {
			result.Error("have no connected to mysql!");
			this.Error(result.ErrorMessage());
			return result;
		}

		this.lock();
		// ------------------------

		while (response == null) {
			ires = this.open_index(dbName, tableName, indexName,
					this.join_string(columns, ","));
			if (ires.Error() != null) {
				result.Error(ires.ErrorMessage());
				this.Error(ires.ErrorMessage());
			}

			// ------------------------
			String cmd = String.format("%d\t%s\t%d\t%s\t%d\t%d\n", ires.PID(),
					oper, key_num, key_vals, limit, skip);
			this.Stack("query cmd:[" + cmd + "]");
			response = this.run_command(cmd);
			if (response == null) {
				this.reconnect();
				continue;
			}
			if (response.result_code != 0) {
				result.Set("Error Something:" + response.data[1].trim(), null);
				this.Error(result.ErrorMessage());
			} else {
				result.Rows(this.parse_query_result(ires.PID(), response));
				this.Debug("Query returned:", result.Rows().size(), "rows");
				if (this.log.GetLevel() >= ILog.Debug2Lv) {
					String info = "\n";
					info += this.join_string(columns, "\t");
					info += "\n--------------------------------------------------------------------------\n";

					for (int i = 0; i < result.Rows().size(); i++) {
						HandlerSocketRow row = result.Rows().get(i);
						String[] row_field = new String[columns.length];// make([]string,
																		// len(columns));
						for (int idx = 0; idx < columns.length; idx++) {
							row_field[idx] = row.GetField_String(columns[idx]);
						}
						info += this.join_string(row_field, "\t") + "\n";
					}
					this.Debug(info);
				}
			}// else

			Core.Gosched();
		}// while
			// ------------------------
		this.unlock();
		this.Debug("Query finish, err:", result.ErrorMessage());
		return result;
	}

	@Override
	public QueryOneResult QueryOne(String dbName, String tableName,
			String indexName, String oper, Object[] keys, String[] columns,
			int skip) {
		this.Debug("QueryOne called");

		IndexResult ires = null;
		CommandResponse response = null;
		QueryOneResult result = new QueryOneResult();
		int key_num = keys.length;
		String key_vals = this.join_hs_field(keys);
		int limit = 1;

		if (!this.conn.IsConnected()) {
			result.Error("have no connected to mysql!");
			this.Error(result.ErrorMessage());
			return result;
		}

		this.lock();
		// ------------------------

		while (response == null) {
			ires = this.open_index(dbName, tableName, indexName,
					this.join_string(columns, ","));

			if (ires.Error() != null) {
				result.Error(ires.ErrorMessage());
				this.Error(ires.ErrorMessage());
			}

			String cmd = String.format("%d\t%s\t%d\t%s\t%d\t%d\n", ires.PID(),
					oper, key_num, key_vals, limit, skip);
			this.Stack("query cmd:[" + cmd + "]");
			response = this.run_command(cmd);
			if (response == null) {
				this.reconnect();
				continue;
			}

			if (response.result_code != 0) {
				result.Set("Error Something:" + response.data[1].trim(), null);
				this.Error(result.ErrorMessage());
			} else {
				Vector<HandlerSocketRow> rows = this.parse_query_result(
						ires.PID(), response);
				if (rows.size() >= 1) {
					result.Error(null);
					result.Row(rows.get(0));
				}
				this.Debug("QueryOne returned:", rows.size(), " rows");
				if (this.log.GetLevel() >= ILog.Debug2Lv) {
					String info = "\n";
					info += this.join_string(columns, "\t");
					info += "\n--------------------------------------------------------------------------\n";
					if (result.Row().NotNull()) {
						String[] row_field = new String[columns.length];// make([]string,
																		// len(columns));
						for (int idx = 0; idx < columns.length; idx++) {
							row_field[idx] = result.Row().GetField_String(
									columns[idx]);
						}
						// fmt.Println(strings.Join(row_field, "\t"));
						info += this.join_string(row_field, "\t");
					}
					info += "\n";
					this.Debug(info);
				}
			}// else

			Core.Gosched();
		}// while
			// ------------------------
		this.unlock();
		this.Debug("QueryOne finish, err:", result.ErrorMessage());
		return result;
	}

	@Override
	public Error Close() {
		this.Debug("Close called");
		Error err = null;

		if (this.conn.IsConnected()) {
			// err = new Error(
			// "A connection to a MySQL server is required to use this function");
			// this.Error(err.getMessage());
			// } else {
			this.Debug("Sent quit command to server");

			// Close connection
			this.conn.Close();
			this.Debug("Closed connection to server");

			// Free indexes
			this.pid_dtc_link.clear();
			this.dtc_pid_link.clear();
			this.pid_columns_link.clear();
			this.pid_counter = 0;
		}

		this.Debug("Close finish, err:", err);
		return err;
	}

	private IndexResult open_index(String dbName, String tableName,
			String indexName, String columns) {

		String dtc = dbName + "$" + tableName + "$" + indexName + "$" + columns;
		IndexResult result = new IndexResult(dtc);

		// ------------------------
		if (this.dtc_pid_link.containsKey(dtc)) {
			result.PID(this.dtc_pid_link.get(dtc));
			return result;
		}

		if (this.pid_counter >= this.index_max) {
			result.PID(1);// pid is max limit
		} else {
			result.PID(this.pid_counter);
		}

		this.Debug("open index called: ", result.PID(), "\t", dtc);
		String cmd = String.format("%s\t%d\t%s\t%s\t%s\t%s\n", "P",
				result.PID(), dbName, tableName, indexName, columns);
		this.Stack("open_index cmd:[" + cmd + "]");
		
		CommandResponse response = this.run_command(cmd);
		while(response==null) {
			Core.Gosched();
			this.reconnect();
			response = this.run_command(cmd);
		}
		
		if (response != null) {
			if (response.result_code != 0) {
				result.Error("Error Opening Index:" + response.data[1].trim());
				this.Error("Error Opening Index:" + result.ErrorMessage());
			} else {
				// Update indexes
				this.pid_counter += 1;

				this.pid_dtc_link.put(result.PID(), dtc);
				this.dtc_pid_link.put(dtc, result.PID());
				this.pid_columns_link.put(result.PID(), columns.split(","));

				this.Debug("create index succeed: ", result.PID(), "\t", dtc);
			}
		}

		// ------------------------
		return result;
	}

	private Vector<HandlerSocketRow> parse_query_result(int pid,
			CommandResponse hsp) {
		Vector<HandlerSocketRow> rows = new Vector<HandlerSocketRow>();
		int fieldCount = Integer.parseInt(hsp.data[0]);
		int remainingFields = hsp.data.length - 1;
		if (fieldCount > 0) {
			int rs = remainingFields / fieldCount;
			// rows = new HandlerSocketRow[rs];

			int offset = 1;

			for (int row_idx = 0; row_idx < rs; row_idx++) {
				Map<String, String> row_data = new HashMap<String, String>();// make(map[string]string,
				// fieldCount)
				for (int fld_idx = 0; fld_idx < fieldCount; fld_idx++) {

					// 在这里转义效率太低，因为很多字段都不是string的，还会破坏非string字段的完整性
					// ff := strings.Replace(hsp.data[offset+fld_idx], TMTab,
					// Tab, -1)
					// ff = strings.Replace(ff, TMReturn, Return, -1)
					// row_data[this.query_indexes[pid][fld_idx]] = ff

					row_data.put(this.pid_columns_link.get(pid)[fld_idx],
							hsp.data[offset + fld_idx]);
				}
				rows.add(new HandlerSocketRow(row_data));// [row_idx]
				offset += fieldCount;
			}
		}
		return rows;
	}

	private CommandResponse run_command(String cmd) {
		LinePacket sent = new LinePacket(512);
		sent.WriteBytes(cmd);

		CommandResponse response = null;
		LinePacket recv = conn.SendCommand(sent);

		// parse
		if (recv != null)
			response = new CommandResponse(recv);
		else
			this.Error("disconnected!");
		return response;
	}

	private Error auth() {
		Error err = null;
		if (!this.conn.IsConnected()) {
			err = new Error("have no connected to mysql!");
			this.Error(err);
			return err;
		}
		CommandResponse response = null;
		if (!this.port_passwd.equals("")) {
			this.Debug("Auth called, passwd:", this.port_passwd);

			String cmd = String
					.format("%s\t%d\t%s\n", "A", 1, this.port_passwd);
			this.Stack("auth cmd:[" + cmd + "]");
			while (response == null) {
				response = this.run_command(cmd);
				if (response == null) {
					this.reconnect();
					continue;
				}
				if (response.result_code != 0) {
					err = new Error("Error Auth:" + response.data[1].trim());
					this.Error(err);
				} else {
					this.Success("Auth succeed!");
				}
			}

		}

		return err;
	}

	private Error connect() {
		this.Debug("Connect called, host:", this.host, " port:", this.host);
		Error err = this.conn.Connect(this.host, this.port, 5000);
		if (err == null) {
			this.pid_dtc_link.clear();
			this.dtc_pid_link.clear();
			this.pid_columns_link.clear();
			this.pid_counter = 0;
			// this.last_op_time_r = new Date().getTime();
			this.Success("Connect succeed!");
			err = this.auth();
		} else {
			err = new Error("Error Connect:" + err.getMessage());
			this.Error(err);
		}
		return err;
	}

	private Error reconnect() {
		this.Debug("Reconnect called");
		this.Close();
		Core.Gosched();
		this.connect();
		return null;
	}

	private String join_string(String[] columns, String flag) {
		String result = "";
		if (columns != null) {
			for (int i = 0; i < columns.length; i++) {
				if (i != columns.length - 1)
					result += columns[i] + flag;
				else
					result += columns[i];
			}
		}

		return result;
	}

	private String join_hs_field(Object... vals) {
		String result = "";
		for (int i = 0; i < vals.length; i++) {
			if (i != vals.length - 1)
				result += vals[i] + "\t";
			else
				result += vals[i];
		}

		return result;
	}

	private void lock() {
		if (this.thread_safe) {
			try {
				this.mutex.acquire();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	private void unlock() {
		if (this.thread_safe) {
			this.mutex.release();
		}
	}

	public void Error(Object... args) {
		if (this.log != null)
			this.log.Error(this.name, args);
	}

	public void Warning(Object... args) {
		if (this.log != null)
			this.log.Warning(this.name, args);
	}

	public void Success(Object... args) {
		if (this.log != null)
			this.log.Success(this.name, args);
	}

	public void Debug(Object... args) {
		if (this.log != null)
			this.log.Debug(this.name, args);
	}

	public void Stack(Object... args) {
		if (this.log != null)
			this.log.Stack(this.name, args);
	}

	public void Temp(Object... args) {
		if (this.log != null)
			this.log.Temp(this.name, args);
	}
}
